package com.bw.yk08;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Collections;

public class ADSAsyncFunction extends RichAsyncFunction<TradeOrdersStats,TradeOrdersStats> {

    private Connection conn1;
    private Connection conn2;

    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn1= DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181");
        conn2= DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181");
    }

    @Override
    public void close() throws Exception {
        if(conn1!=null)
            conn1.close();
        if(conn2!=null)
            conn2.close();
    }

    @Override
    public void asyncInvoke(TradeOrdersStats tradeOrdersStats, ResultFuture<TradeOrdersStats> resultFuture) throws Exception {

        String sql1="select line from dim8_product_spu where id=?";
        PreparedStatement ps1=null;
        try{
            ps1=conn1.prepareStatement(sql1);
            ps1.setObject(1,tradeOrdersStats.getProduct_spu_id());
            java.sql.ResultSet rs1=ps1.executeQuery();
            if(rs1.next())
            {
                String line1=rs1.getString("line");
                if(line1!=null)
                {
                    JSONObject ob1= JSON.parseObject(line1);
                    String spu_name=ob1.getString("name");
                    tradeOrdersStats.setSpu_name(spu_name);
                }
            }
        }catch (Exception e)
        {
            e.printStackTrace();
        }

        String sql2="select line from dim8_region where id=?";
        PreparedStatement ps2=null;
        try{
            ps2=conn2.prepareStatement(sql1);
            ps2.setObject(1,tradeOrdersStats.getRegion_code());
            java.sql.ResultSet rs2=ps2.executeQuery();
            if(rs2.next())
            {
                String line2=rs2.getString("line");
                if(line2!=null)
                {
                    JSONObject ob1= JSON.parseObject(line2);
                    String region_name=ob1.getString("name");
                    tradeOrdersStats.setRegion_name(region_name);
                }
            }
        }catch (Exception e)
        {
            e.printStackTrace();
        }



        resultFuture.complete(Collections.singleton(tradeOrdersStats));

    }

}
